package dbMysql

import (
	"fmt"
	. "gitee.com/kingzyt/common/idxRangeAllocer"
	"gitee.com/kingzyt/common/log"
	. "gitee.com/kingzyt/common/util"
	"reflect"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

/*
以插入数据的顺序作为唯一确定的数据先后顺序, 数据中的时间RecordTimeNS仅仅作为查询时的参考,
因为时间很可能因为主机间时间同步或是手动的修改而发生先后变动.
这里不严格要求每一个表的数据总量必须刚刚好为oneTableMaxCnt,
通过后台定时刷新的goroutine来保证能在较小的数量偏差内做好表的切换,
同时保证在索取querySql时只有读锁, 保证并发效率,
允许数量误差就能降低设计的复杂度
*/

const g_table_name_mark = "$tableName$"
const G_TableDef_PKWithAI_Id_and_Index_RecordTimeNS = `
	Id bigint not null auto_increment,
	RecordTimeNS bigint not null,
	primary key(Id),
	index(RecordTimeNS),
`

type ShardingByCount struct {
	name           string
	tableNameMain  string
	oneTableMaxCnt int32

	mutex   sync.RWMutex
	sqlRWer *SimpleMySqlRWer

	curTableIdx     int32
	curTableName    string
	curTableDataCnt AtomicInt32

	idxTimeRanges []*IdxRange

	startTableIdx     int32
	startTableDataCnt int32

	query           *ShardingQueryMulit
	dbTableSwitchOk bool
}

func NewShardingByCount(name string, sqlRWer *SimpleMySqlRWer, tableNameMain string, oneTableMaxCnt int32, createTableTemplate string) (*ShardingByCount, error) {
	if sqlRWer == nil {
		return nil, fmt.Errorf("%s; NewShardingByCount fail, sqlRWer is nil", name)
	}
	if oneTableMaxCnt <= 0 {
		return nil, fmt.Errorf("%s; NewShardingByCount fail, oneTableMaxCnt should be >= 0", name)
	}
	if !strings.Contains(createTableTemplate, g_table_name_mark) {
		return nil, fmt.Errorf("%s; NewShardingByCount fail, createTableTemplate should contain $tableName$ to mark table name", name)
	}
	if !strings.Contains(createTableTemplate, G_TableDef_PKWithAI_Id_and_Index_RecordTimeNS) {
		return nil, fmt.Errorf("%s; NewShardingByCount fail, createTableTemplate should contain G_TableDef_PKWithAI_Id_and_Index_RecordTimeNS to def basic column", name)
	}
	log.Info(name, log.I, "ShardingByCount, allow table row count max: %d", oneTableMaxCnt)
	return &ShardingByCount{
		name:           name,
		tableNameMain:  tableNameMain,
		oneTableMaxCnt: oneTableMaxCnt,
		sqlRWer:        sqlRWer,
		idxTimeRanges:  []*IdxRange{},
		query:          &ShardingQueryMulit{queryUnits: map[int32]*ShardingQueryUnit{}, createTableTemplate: createTableTemplate},
	}, nil
}

func (self *ShardingByCount) getTableName(idx int32) string {
	return fmt.Sprintf("%s_%d", self.tableNameMain, idx)
}

func (self *ShardingByCount) Start() (ok bool) {
	self.mutex.Lock()
	defer self.mutex.Unlock()

	defer func() {
		if ok {
			self.bgCheckSwitchShardingTable()
		}
	}()

	queryDescs, err := GetMysqlTableDescriptions(self.name, self.tableNameMain+"_%", self.sqlRWer)
	if err != nil {
		log.Error(self.name, "ShardingByCount get tables descriptions fail, tableNameMain: %s, err: %s", self.tableNameMain, err)
		return false
	}
	if len(queryDescs) <= 0 {
		ok = self.switchTable()
		return
	}

	tableIdxs := []int32{}

	tableNameRegexp := regexp.MustCompile(".*_(\\d+)$")
	for i := 0; i < len(queryDescs); i++ {
		qd := queryDescs[i]
		matches := tableNameRegexp.FindStringSubmatch(qd.TableName)
		if len(matches) != 1+1 {
			log.Error(self.name, "ShardingByCount table %s name is invalid, should be like xxxx_9", qd.TableName)
			return false
		}

		idx, _ := strconv.Atoi(matches[1])
		tableIdxs = append(tableIdxs, int32(idx))
	}

	sort.Sort(Int32List(tableIdxs))
	self.curTableIdx = tableIdxs[len(tableIdxs)-1]
	self.curTableName = self.getTableName(self.curTableIdx)

	//fmt.Printf("tableIdxs:%v\n", tableIdxs)

	compareBase := queryDescs[0]
	idFound := false
	tnsFound := false
	for i := 0; i < len(compareBase.ColumnNames); i++ {
		if compareBase.ColumnNames[i] == "Id" && compareBase.ColumnTypes[i] == reflect.Int64 && compareBase.ColumnAutoIncrements[i] {
			idFound = true
		} else if compareBase.ColumnNames[i] == "RecordTimeNS" && compareBase.ColumnTypes[i] == reflect.Int64 {
			tnsFound = true
		}
	}
	if !idFound || !tnsFound {
		log.Error(self.name, "ShardingByCount table %s struct is invalid, should contain bigint 'Id' & bigint 'RecordTimeNS'", compareBase.TableName)
		return false
	}

	if !CheckTablesDescriptionSame(self.name+"_ShardingByCount", queryDescs) {
		return false
	}

	for i := 0; i < len(tableIdxs); i++ {
		idx := tableIdxs[i]

		tableName := self.getTableName(idx)

		t0 := int64(0)
		t1 := int64(0)
		err := self.sqlRWer.ReadQueryRow(
			fmt.Sprintf("select ifnull((select RecordTimeNS from %s order by RecordTimeNS asc limit 1),0), ifnull((select RecordTimeNS from %s order by RecordTimeNS desc limit 1),0);", tableName, tableName),
			[]interface{}{&t0, &t1}, nil)
		if err != nil {
			log.Error(self.name, "ShardingByCount get table %s min&max RecordTimeNS fail, err:%s", tableName, err)
			return false
		}

		if idx == self.curTableIdx && (t0 == 0 || t1 == 0) {
			t0 = time.Now().UnixNano()
			t1 = t0
		}
		if t0 == 0 || t1 == 0 {
			continue
		}

		self.idxTimeRanges = append(self.idxTimeRanges, &IdxRange{Id: int64(idx), Down: t0, Up: t1})
	}

	SortIdxRange(self.idxTimeRanges)

	_, err = CheckRangeCross(self.idxTimeRanges)
	if err != nil {
		// 因为服务器时间修正或是测试性修改,导致时间点的连续性被破坏, 这里不阻止这种情况, 仅仅是提示可能在查询时出现问题
		log.Warn(self.name, "ShardingByCount some table time range cross, may cause unexpected query results, err: %s", err)
	}
	// fmt.Printf("idxTimeRanges: %v\n", self.idxTimeRanges)

	// 虽然auto_increment是增加和修改都递增的, 不过这里仅仅是insert,所以可以用最大最小的id差来替代count(*), 以提高大表查询效率
	cnt := int32(0)
	err = self.sqlRWer.ReadQueryRow(
		fmt.Sprintf("select ifnull((select Id from %s order by Id desc limit 1),-1) - ifnull((select Id from %s order by Id asc limit 1),0) + 1;", self.curTableName, self.curTableName),
		[]interface{}{&cnt}, nil)
	if err != nil {
		log.Error(self.name, "ShardingByCount get dataCnt from table %s fail, err:%s", self.curTableName, err)
		return false
	}
	self.curTableDataCnt.Set(cnt)
	for _, q := range self.query.queryUnits {
		q.queryCurrent = strings.Replace(q.queryTemplate, g_table_name_mark, self.curTableName, -1)
	}
	self.dbTableSwitchOk = true

	self.startTableIdx = self.curTableIdx
	self.startTableDataCnt = cnt

	log.Info(self.name, log.I, "current table %s", self.curTableName)

	ok = true
	return
}

var g_insertSqlRegexp = regexp.MustCompile(".*\\s*insert\\s*.*")

func (self *ShardingByCount) AddQueryDef(queryId int32, args ...interface{}) {
	if len(args) != 1 {
		panic(fmt.Errorf("%s; parameters should be queryId, queryTemplate(string)", self.name))
	}
	queryTemplate, ok := args[0].(string)
	if !ok {
		panic(fmt.Errorf("%s; args[0] queryTemplate should be string", self.name))
	}

	if !strings.Contains(queryTemplate, g_table_name_mark) {
		panic(fmt.Errorf("%s; AddInsertQueryDef fail, queryId %d, queryTemplate should contain %s to mark table name", self.name, queryId, g_table_name_mark))
	}
	if !g_insertSqlRegexp.MatchString(queryTemplate) {
		panic(fmt.Errorf("%s; query queryId %d is not a insert sql", self.name, queryId))
	}

	self.mutex.Lock()
	defer self.mutex.Unlock()

	_, existed := self.query.queryUnits[queryId]
	if existed {
		panic(fmt.Errorf("query queryId %d has already been set", queryId))
	}

	self.query.queryUnits[queryId] = &ShardingQueryUnit{queryTemplate: queryTemplate}
}

func (self *ShardingByCount) GetQuerySqlStr(queryId int32, _ int64) (querySql string, ok bool) {
	self.mutex.RLock()
	defer self.mutex.RUnlock()

	if !self.dbTableSwitchOk {
		return "", false
	}

	q, ok2 := self.query.queryUnits[queryId]
	if !ok2 || q == nil {
		log.Error(self.name, "queryId(%d) is invalid", queryId)
		return "", false
	}

	return q.queryCurrent, true
}

/*
虽然有将计数累加交由外部控制是不太稳妥, 但根据语句执行的方式,比如单句或是批量等, 这里无法作出合理的预测,
所以妥协为当成功insert数据时由外部累加, 可能会有计数比实际数据量小的情况,
而且在切换表的时候, 可能会有少部分在上一个表GetInsertDataQuerySql, 但在切换后OnInsertDataOk的计数会被计入新表的数量中
*/
func (self *ShardingByCount) OnQueryOk(cnt int32) {
	self.curTableDataCnt.Add(cnt)
}

// 切换表对于外部使用而言透明, 数据不应该严格要求一定要写入哪一个表, 主要能保证在一个可接受的误差数据行数的范围内, 将数据划分到若干表中
func (self *ShardingByCount) switchTable() bool {
	createTableSql := strings.Replace(self.query.createTableTemplate, g_table_name_mark, self.getTableName(self.curTableIdx+1), -1)
	_, err := self.sqlRWer.WriteQuery(createTableSql)
	if err != nil {
		log.Error(self.name, "ShardingByCount switchShardingTable fail, err: %s", err)
	} else {
		nowTNS := time.Now().UnixNano()
		if len(self.idxTimeRanges) > 0 {
			if self.startTableIdx == self.curTableIdx && self.startTableDataCnt == self.curTableDataCnt.Get() {
				// do not modify no changed old table
			} else {
				lastTimeRange := self.idxTimeRanges[len(self.idxTimeRanges)-1]
				if lastTimeRange.Up < nowTNS {
					lastTimeRange.Up = nowTNS
				}
			}
		}

		self.curTableIdx++
		self.curTableName = self.getTableName(self.curTableIdx)
		self.curTableDataCnt.Set(0)
		for _, q := range self.query.queryUnits {
			q.queryCurrent = strings.Replace(q.queryTemplate, g_table_name_mark, self.curTableName, -1)
		}
		self.idxTimeRanges = append(self.idxTimeRanges, &IdxRange{Id: int64(self.curTableIdx), Down: nowTNS, Up: nowTNS})
		log.Info(self.name, log.I, "switch table to %s", self.curTableName)
		//fmt.Printf("idxTimeRanges: %v\n", self.idxTimeRanges)
	}
	self.dbTableSwitchOk = err == nil
	return self.dbTableSwitchOk
}

func (self *ShardingByCount) bgCheckSwitchShardingTable() {
	go UpdateProc(time.Second*10, true, func() (ok bool) {
		// oneTableMaxCnt never change
		if self.curTableDataCnt.Get() < self.oneTableMaxCnt {
			return true
		}

		self.mutex.Lock()
		defer self.mutex.Unlock()
		self.switchTable()
		return true
	})
}

func (self *ShardingByCount) GetTableNamesInTimeRange(beginT int64, endT int64) (tableNames []string) {
	self.mutex.RLock()
	defer self.mutex.RUnlock()

	for i := 0; i < len(self.idxTimeRanges); i++ {
		r := self.idxTimeRanges[i]
		up := r.Up
		if i == len(self.idxTimeRanges)-1 {
			// 最后一个idxRange的Up只在下一次切换表格的时候被更新, 所以实际上就是指当前时间
			up = time.Now().UnixNano()
		}
		if r.Down <= endT && up >= beginT {
			tableNames = append(tableNames, self.getTableName(int32(r.Id)))
		}
	}
	return
}

func (self *ShardingByCount) GetSqlRWer() *SimpleMySqlRWer {
	return self.sqlRWer
}
